package com.lmf.cloud.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;

/**
 * @Description kafka生产者
 * <p>
 * Copyright @2022 Shanghai Mise Co. Ltd.
 * All right reserved.
 * <p>
 * @Author LiuMingFu
 * @Date 2024/8/3 11:24
 */
public class KafkaProducerMain {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //1.配置属性集合
        Map<String, Object> configMap = new HashMap<>();
        //kafka 集群地址
        configMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        //kafka中 key 序列化配置
//        configMap.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        configMap.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //kafka中 value 序列化配置
//        configMap.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        configMap.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //2.创建kafka生产者客户端：建立Kafka连接，构造对象时，需要传递配置参数
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(configMap);

        //循环多次
        for (int i = 0; i < 10; i++) {
            //3.准备数据：构造对象时需要传递 【Topic主题名称】，【Key】，【Value】三个参数
            ProducerRecord<String, String> record = new ProducerRecord<>("depart", 0, "k" + i, "hello kafka " + i);
            ProducerRecord<String, String> record2 = new ProducerRecord<>("depart", 1, "k2" + i, "hello kafka " + i);
            ProducerRecord<String, String> record3 = new ProducerRecord<>("depart", 2, "k3" + i, "hello kafka " + i);

            //4.生产者发送数据——异步发送数据
            kafkaProducer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    System.out.println("发送数据失败：" + exception.getMessage());
                } else {
                    System.out.println("发送数据成功：" + metadata.toString());
                }
            });
            kafkaProducer.send(record2, (metadata, exception) -> {
                if (exception != null) {
                    System.out.println("发送数据失败：" + exception.getMessage());
                } else {
                    System.out.println("发送数据成功：" + metadata.toString());
                }
            });
            kafkaProducer.send(record3, (metadata, exception) -> {
                if (exception != null) {
                    System.out.println("发送数据失败：" + exception.getMessage());
                } else {
                    System.out.println("发送数据成功：" + metadata.toString());
                }
            });


            //5.生产者发送数据——同步发送数据
//            kafkaProducer.send(record, (metadata, exception) -> {
//                if (exception != null) {
//                    System.out.println("发送数据失败：" + exception.getMessage());
//                } else {
//                    System.out.println("发送数据成功：" + metadata.toString());
//                }
//            }).get();
        }


        //5.关闭生产者客户端：断开与kafka连接
        kafkaProducer.close();
    }
}
